// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-12-05

package src

import (
	"context"
	"fmt"
	"gitee.com/go-libs/log"
	"gitee.com/go-libs/process"
	"gitee.com/go-libs/runtime"
	"gitee.com/gomq/sr"
	amqp "github.com/rabbitmq/amqp091-go"
	"net/http"
	"strconv"
	"sync/atomic"
	"time"
)

type consumer struct {
	executor     sr.Executor
	manager      *manager
	subscription sr.Subscription

	initializer bool
	name        string
	process     process.Process
	receiving   int32
	restarts    int64

	delayExchange sr.TopicName
	queue         sr.QueueName
}

// + Events.

func (o *consumer) OnAcknowledge(ctx context.Context, delivery amqp.Delivery, requeue bool) {
	runtime.GetCounter().MqCounter().IncrAcknowledgedCount(1)

	// Requeue message if error occurred on republish.
	if requeue {
		if err := delivery.Nack(false, requeue); err != nil {
			log.Warnfc(ctx, `[servlet=rabbitmq][%s] message requeue: error="%v"`, o.name, err)
			return
		}
		log.Infofc(ctx, `[servlet=rabbitmq][%s] message requeue`, o.name)
		return
	}

	// Acknowledge received message, It means the message will be deleted
	// from queue
	if err := delivery.Ack(false); err != nil {
		log.Warnfc(ctx, `[servlet=rabbitmq][%s] message acknowledge: error="%v"`, o.name, err)
		return
	}
	log.Infofc(ctx, `[servlet=rabbitmq][%s] message acknowledged`, o.name)
}

func (o *consumer) OnDelivery(conn *amqp.Connection, delivery amqp.Delivery) {
	var (
		delayMilliseconds int64
		header            = o.getHeaders(delivery)
		msg               = sr.NewMessageWithHeader(header)

		ct      = time.Now()
		ctx     = msg.Ctx
		requeue = false
	)

	// Clean when delivery done.
	defer func() {
		// Recover and catch runtime panic if necessary.
		if r := recover(); r != nil {
			log.Fatalfc(ctx, `[servlet=rabbitmq][%s] receiving caused an panic: %v`, o.name, r)
		}

		// Acknowledge or requeue the message.
		o.OnAcknowledge(ctx, delivery, requeue)

		// Release message into pool.
		if msg != nil {
			msg.Release()
		}
	}()

	// Parse received delivery into message fields.
	msg.ConsumedCount, msg.ProducedCount, msg.MessageId, msg.Time = o.getProperties(delivery, ct)
	msg.Body, msg.Delay = delivery.Body, ct.Sub(msg.Time)
	msg.Topic, msg.Tag, msg.Queue = o.subscription.GetTopic(), o.subscription.GetTag(), o.queue

	// Ignore 1: Servlet executor to process consumption not defined.
	if o.executor == nil {
		log.Warnfc(ctx, `[servlet=rabbitmq][%s] discard message: for="executor not defined", %s`, o.name, msg)
		return
	}

	// Delay by header.
	if v, ok := delivery.Headers[headerDelaySeconds]; ok {
		delete(delivery.Headers, headerDelaySeconds)
		if vn, ve := strconv.ParseInt(fmt.Sprintf(`%v`, v), 10, 64); ve == nil && vn > 0 {
			delayMilliseconds = vn
		}
	}

	// Republish for delay subscription.
	if delay := o.subscription.GetDelayMilliseconds() + delayMilliseconds; delay > 0 {
		if ms := delay - msg.Delay.Milliseconds(); ms >= 10 {
			cts := log.NewContextForChildInfo(ctx, `[servlet=rabbitmq][%s] republish message: for="delay subscription", message=%s`, o.name, msg)
			dur := time.Duration(ms) * time.Millisecond
			requeue = o.OnRepublish(cts, conn, delivery, msg, dur) != nil
			return
		}
	}

	// Ignore 2: too many consumed count.
	if msg.ConsumedCount += 1; msg.ConsumedCount > o.subscription.GetMaxRetry() {
		log.Infofc(ctx, `[servlet=rabbitmq][%s] discard message: for="too many consume count", %s`, o.name, msg)
		return
	}

	// Call executor to invoke consumption, Return if nil returned.
	runtime.GetCounter().MqCounter().IncrScheduledCount(1)
	log.Infofc(ctx, `[servlet=rabbitmq][%s] message received: %s`, o.name, msg)
	if err := o.executor(o.subscription, msg); err == nil {
		runtime.GetCounter().MqCounter().IncrSuccessScheduledCount(1)
		log.Infofc(ctx, `[servlet=rabbitmq][%s] discard message: for="success scheduled"`, o.name)
		return
	}

	// Ignore 3: for too many failed consumptions.
	runtime.GetCounter().MqCounter().IncrFailScheduledCount(1)
	if msg.ConsumedCount >= o.subscription.GetMaxRetry() {
		log.Infofc(ctx, `[servlet=rabbitmq][%s] discard message: for="too many fails"`, o.name)
		return
	}

	// Republish for retry later.
	cts := log.NewContextForChildInfo(ctx, `[servlet=rabbitmq][%s] republish message: for="retry later"`, o.name)
	dur := time.Duration(msg.ConsumedCount) * time.Minute
	requeue = o.OnRepublish(cts, conn, delivery, msg, dur) != nil
}

func (o *consumer) OnRepublish(ctx context.Context, conn *amqp.Connection, delivery amqp.Delivery, msg *sr.Message, dur time.Duration) (err error) {
	var (
		channel    *amqp.Channel
		publishing = o.manager.client.GeneratePublishing()
	)

	runtime.GetCounter().MqCounter().IncrProducedCount(1)

	// Clean when republish process done.
	defer func() {
		// Recover and catch runtime panic.
		if r := recover(); r != nil {
			log.Fatalfc(ctx, `[servlet=rabbitmq][%s] republishing caused an panic: %v`, o.name, r)

			// Set returned error if not defined.
			if err == nil {
				err = sr.NewError(http.StatusInternalServerError, `%v`, r)
			}
		}

		// Close opened channel.
		o.manager.client.CloseChannel(ctx, channel)
	}()

	// Generate publishing fields.
	publishing.Body = delivery.Body
	publishing.Expiration = fmt.Sprintf(`%d`, dur.Milliseconds())
	publishing.Headers = delivery.Headers
	publishing.MessageId = msg.MessageId
	publishing.Timestamp = msg.Time

	// Publishing counters.
	publishing.Headers[headerConsumedCount] = msg.ConsumedCount
	publishing.Headers[headerProducedCount] = msg.ProducedCount + 1
	publishing.Headers[headerProducedTime] = msg.Time.UnixMicro()

	// Open channel with given connection
	if channel, err = o.manager.client.OpenChannel(ctx, conn); err != nil {
		return
	}

	// Call publisher.
	err = o.manager.client.Publish(ctx, channel, o.delayExchange, o.subscription.GetTag(), publishing)
	return
}

func (o *consumer) Process() process.Process {
	return o.process
}

// + Process

func (o *consumer) After(_ context.Context) (err error) {
	if log.Config().DebugOn() {
		log.Debugf(`[servlet=rabbitmq][%s] stopped`, o.name)
	}
	return
}

func (o *consumer) Before(_ context.Context) (err error) {
	if log.Config().DebugOn() {
		log.Debugf(`[servlet=rabbitmq][%s] startup`, o.name)
	}
	return
}

func (o *consumer) Run(ctx context.Context) (err error) {
	// Initialize
	// for first runner.
	if o.initializer {
		if err = o.manager.remoting.Initialize(ctx, o.subscription); err != nil {
			o.sleeper(ctx)
			return
		}
		o.initializer = false
		atomic.StoreInt64(&o.restarts, 0)
	}

	// Increment
	// receive counter.
	atomic.AddInt32(&o.receiving, 1)

	// Receive
	// queue message in a goroutine.
	go func() {
		// Decrement
		// receive counter.
		defer atomic.AddInt32(&o.receiving, -1)

		// Loop receiver until context cancelled.
		for {
			if ctx == nil || ctx.Err() != nil {
				return
			}

			o.receive(ctx)

			// Too many redo receiver for connection closed.
			o.sleeper(ctx)
		}
	}()

	// Receive
	// channel message of the context.
	for {
		select {
		case <-ctx.Done():
			for {
				if atomic.LoadInt32(&o.receiving) == 0 {
					break
				}
			}
			return
		}
	}
}

// + Access

func (o *consumer) getHeaders(delivery amqp.Delivery) http.Header {
	header := make(http.Header)
	for k, v := range delivery.Headers {
		header.Set(k, fmt.Sprintf(`%v`, v))
	}
	return header
}

func (o *consumer) getProperties(delivery amqp.Delivery, tm time.Time) (consumedCount, producedCount int, messageId string, messageTime time.Time) {
	// Fix: MessageId.
	if messageId = delivery.MessageId; messageId == "" {
		messageId = sr.NewHash().String()
	}

	// Fix: MessageTime.
	messageTime = tm
	if v, ok := delivery.Headers[headerProducedTime]; ok {
		if n, ne := strconv.ParseInt(fmt.Sprintf(`%v`, v), 10, 64); ne == nil && n > 0 {
			if t := time.UnixMicro(n); t.Year() >= 2020 {
				messageTime = t
			}
		}
	}

	// Fix: ConsumedCount.
	consumedCount = 0
	if v, ok := delivery.Headers[headerConsumedCount]; ok {
		if n, ne := strconv.ParseInt(fmt.Sprintf(`%v`, v), 10, 32); ne == nil && n > 0 {
			consumedCount = int(n)
		}
	}

	// Fix: ProducedCount.
	producedCount = 1
	if v, ok := delivery.Headers[headerProducedCount]; ok {
		if n, ne := strconv.ParseInt(fmt.Sprintf(`%v`, v), 10, 32); ne == nil && n > 0 {
			producedCount = int(n)
		}
	}
	return
}

func (o *consumer) init() *consumer {
	o.initializer = true
	o.process = process.New(o.name, o)

	// Initialize exchange name and queue name.
	_, o.delayExchange, _ = o.manager.client.GenerateExchange(o.subscription.GetId(), o.subscription.GetTopic())
	o.queue, _ = o.manager.client.GenerateQueue(o.subscription.GetId())
	return o
}

func (o *consumer) receive(ctx context.Context) {
	var (
		ch         *amqp.Channel
		conn       *amqp.Connection
		delivering int32
		err        error
		list       <-chan amqp.Delivery
	)

	log.Infof(`[servlet=rabbitmq][%s] receiver begin`, o.name)
	defer log.Infof(`[servlet=rabbitmq][%s] receiver finish`, o.name)

	// Clean
	// when receiver done.
	defer func() {
		if r := recover(); r != nil {
			log.Fatalf(`[servlet=rabbitmq][%s] receiving caused an panic: %v`, o.name, r)
		}

		// Loop
		// until delivering process done.
		for {
			if atomic.LoadInt32(&delivering) == 0 {
				break
			}
			time.Sleep(time.Millisecond * 10)
		}

		// Close channel and connection.
		o.manager.client.CloseChannel(ctx, ch)
		o.manager.client.CloseConnection(ctx, conn)
	}()

	// Open connection.
	if conn, err = o.manager.client.OpenConnection(ctx); err != nil {
		return
	}

	// Open channel.
	if ch, err = o.manager.client.OpenChannel(ctx, conn); err != nil {
		return
	}

	// Start consuming.
	if list, err = o.manager.client.Consume(ctx, ch, o.queue, o.subscription.GetConcurrency()); err != nil {
		return
	}

	// Loop
	// received delivery.
	for v := range list {
		// Increment counter.
		atomic.AddInt32(&delivering, 1)

		// Trigger in a goroutine.
		go func(delivery amqp.Delivery) {
			runtime.GetCounter().MqCounter().IncrReceivedCount(1)

			// Fix deliver headers.
			if delivery.Headers == nil {
				delivery.Headers = make(amqp.Table)
			}

			// Trigger event.
			o.OnDelivery(conn, delivery)

			// Decrement counter.
			atomic.AddInt32(&delivering, -1)
		}(v)
	}
}

func (o *consumer) sleeper(ctx context.Context) {
	if ctx.Err() == nil {
		n := atomic.AddInt64(&o.restarts, 1)

		if n > 3 {
			time.Sleep(time.Second * 3)
		}
	}
}
